# -*- coding: utf-8 -*-
"""
项目名称：NoSql_kafka_project
文件名称：kafka_consumer.py.py
创建时间：2025-06-05 16:20:55

系统用户：Administrator
作　　者：無以菱
联系邮箱：huangjing2001.guet@qq.com
功能描述：
- 功能：从Kafka读取订单数据并写入HBase
- 配置：连接到Kafka集群和HBase服务器
- 消费组：order_processor
"""
from kafka import KafkaConsumer  
import json  
import happybase  
import time  
import sys


def run_kafka_consumer():
    """
    运行Kafka消费者，从Kafka读取订单数据并写入HBase
    
    该函数创建一个Kafka消费者实例，连接到HBase，然后持续从Kafka读取消息并写入HBase表
    """
    # 初始化Kafka消费者
    # 参数说明:
    # 'real_time_orders': 要消费的主题名称
    # bootstrap_servers: 指定Kafka集群地址
    # auto_offset_reset='latest': 从最新消息开始消费，而不是从头开始
    # enable_auto_commit=True: 自动提交消费位移
    # group_id: 消费者组ID，相同组的消费者共享消费进度
    # value_deserializer: 定义如何反序列化消息值，这里将JSON字符串解码并转换为Python对象
    consumer = KafkaConsumer(
        'real_time_orders',
        bootstrap_servers=['192.168.10.10:9092', '192.168.10.11:9092', '192.168.10.12:9092'],
        auto_offset_reset='latest',
        enable_auto_commit=True,
        group_id='order_processor',
        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
    )

    # 连接HBase
    try:
        # 创建到HBase服务器的连接
        connection = happybase.Connection('192.168.10.10', port=9090)
        # 获取指定表的引用
        table = connection.table('real_time_orders_table')
        print("已连接到HBase")
    except Exception as e:
        # 如果连接失败，打印错误信息并退出函数
        print(f"连接HBase失败: {e}")
        return

    print("Kafka消费者启动，开始处理订单数据...")

    # 消费消息并写入HBase
    try:
        # 迭代消费消息，这是一个无限循环，会持续等待并处理新消息
        for message in consumer:
            try:
                # 获取消息值（已通过deserializer转换为Python对象）
                order = message.value
                # 创建行键，使用订单ID和创建时间的组合确保唯一性
                row_key = f"{order['ORDER_ID']}_{order['CREATE_TIME']}"

                # 准备要写入HBase的数据
                # HBase要求列族和列名为bytes类型，值为字符串类型
                data = {
                    b'cf:ORDER_ID': str(order['ORDER_ID']),  # 订单ID
                    b'cf:PAY_AMOUNT': str(order['PAY_AMOUNT']),  # 支付金额
                    b'cf:PAYWAY': order['PAYWAY'],  # 支付方式
                    b'cf:CREATE_TIME': str(order['CREATE_TIME'])  # 创建时间
                }

                # 写入HBase表，行键也需要编码为bytes类型
                table.put(row_key.encode('utf-8'), data)
                print(f"订单数据已写入HBase: {order}")

            except Exception as e:
                # 处理单条消息时的错误，继续处理下一条
                print(f"处理消息时发生错误: {e}")
    except KeyboardInterrupt:
        # 捕获键盘中断信号（Ctrl+C），优雅地退出
        print("Kafka消费者已停止")
    finally:
        # 确保在任何情况下都关闭HBase连接
        connection.close()
        print("HBase连接已关闭")

if __name__ == "__main__":
    # 单独运行此文件时执行
    print("正在启动Kafka消费者进行调试...")
    try:
        run_kafka_consumer()  # 启动Kafka消费者
    except KeyboardInterrupt:
        print("Kafka消费者已停止")  # 处理键盘中断
    except Exception as e:
        print(f"Kafka消费者运行出错: {e}")  # 处理其他异常
        sys.exit(1)  # 非正常退出，返回错误码1